package com.cctc.rds.nrpc.core.client;

import com.cctc.rds.nrpc.api.client.RpcClient;
import com.cctc.rds.nrpc.api.client.config.RpcClientConfig;
import com.cctc.rds.nrpc.api.connection.client.ClientConnectionListener;
import com.cctc.rds.nrpc.api.connection.client.ClientConnectionListenerProvider;
import com.cctc.rds.nrpc.api.exception.ConnectionTimeoutException;
import com.cctc.rds.nrpc.api.exception.RpcIOException;
import com.cctc.rds.nrpc.api.exception.WaitResponseTimeoutException;
import com.cctc.rds.nrpc.api.rpc.RpcRequest;
import com.cctc.rds.nrpc.api.rpc.RpcResponse;
import com.cctc.rds.nrpc.core.client.connection.ClientConnection;
import com.cctc.rds.nrpc.core.client.connection.ClientConnectionStatus;
import com.cctc.rds.nrpc.core.client.handler.ClientIdleStateHandler;
import com.cctc.rds.nrpc.core.client.handler.ClientRpcMessageHandler;
import com.cctc.rds.nrpc.core.client.helper.ChannelPromiseHolder;
import com.cctc.rds.nrpc.core.client.helper.TxidGenetor;
import com.cctc.rds.nrpc.core.protocol.codec.MessageDecode;
import com.cctc.rds.nrpc.core.protocol.codec.MessageEncode;
import com.cctc.rds.nrpc.core.protocol.codec.MessageUnpacket;
import com.cctc.rds.nrpc.core.protocol.message.Message;
import com.cctc.rds.nrpc.core.protocol.message.impl.RpcRequestMessage;
import com.cctc.rds.nrpc.core.util.IdUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class NettyRpcClient implements RpcClient {
    private NioEventLoopGroup workerGroup;
    private RpcClientConfig config;
    private ClientConnection connection;
    private ClientRpcMessageHandler rpcMessageHandler;
    private final AtomicInteger status = new AtomicInteger();
    private final AtomicBoolean shouldRetry = new AtomicBoolean();
    private final AtomicLong failRetryConnectMs = new AtomicLong();
    private final ScheduledExecutorService failRetryConnectScheduler
            = new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "client-recon-scheduler-" + IdUtils.newId()));
    private final Runnable failRetryConnectCallable = () -> {
        if (!shouldRetry.get()) return;
        failRetryConnectScheduler.schedule(() -> {
            if (shouldRetry.get()) {
                start();
            }
        }, failRetryConnectMs.get(), TimeUnit.MILLISECONDS);
        clear();
    };

    @Override
    public void start(RpcClientConfig config) {
        this.config = config;
        this.status.set(ClientConnectionStatus.OFF.getValue());
        this.failRetryConnectMs.set(config.getFailRetryConnectMs());
        this.shouldRetry.set(Boolean.TRUE);
        start();
    }

    private void start() {
        Bootstrap bootstrap = new Bootstrap();
        workerGroup = new NioEventLoopGroup();
        bootstrap
                .group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.SO_SNDBUF, 32 * 1024)
                .option(ChannelOption.SO_RCVBUF, 32 * 1024)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) config.getConnectionTimeoutMs())
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new MessageUnpacket())
                                .addLast(new MessageDecode())
                                .addLast(new MessageEncode())
                                .addLast(new ClientIdleStateHandler(config.getIdleStateMs()))
                                .addLast(new ClientRpcMessageHandler(status, failRetryConnectCallable));
                    }
                });
        if (!status.compareAndSet(ClientConnectionStatus.OFF.getValue(), ClientConnectionStatus.PENDING.getValue()))
            return;
        bootstrap
                .connect(config.getRemoteIp(), config.getRemotePort())
                .addListener((GenericFutureListener<ChannelFuture>) future -> {
                    if (future.cause() != null) {
                        if (connection != null) {
                            try {
                                connection.close();
                            } catch (Exception ignored) {
                            } finally {
                                connection = null;
                            }
                        }
                        List<ClientConnectionListener> listeners = ClientConnectionListenerProvider.getInstance().list();
                        if (listeners != null && listeners.size() > 0) {
                            listeners.stream().forEach(listener -> listener.onTimeout(config.getRemoteIp(), config.getRemotePort()));
                        }
                    } else {
                        connection = new ClientConnection(future.channel());
                    }
                });

    }

    @Override
    public boolean isReady() {
        return connection != null
                && rpcMessageHandler != null
                && rpcMessageHandler.hasStarted();
    }

    @Override
    public RpcResponse remoteInvoke(RpcRequest rpcRequest) throws RpcIOException {
        if (!isReady()) throw new ConnectionTimeoutException("连接异常！！！");
        String txid = TxidGenetor.genTxid();
        try {
            RpcRequestMessage rpcRequestMessage = RpcRequestMessage.builder().txid(txid).rpcRequest(rpcRequest).build();
            connection.getChannel().writeAndFlush(rpcRequestMessage);
            ChannelPromise promise = connection.getChannel().newPromise();
            ChannelPromiseHolder.putEntry(txid, promise);
            if (!promise.await(config.getWaitResponseTimeoutMs())) {
                throw new WaitResponseTimeoutException("等待响应超时！！！");
            }
            Map.Entry<ChannelPromise, Message> entry = ChannelPromiseHolder.getEntry(txid);
            if (entry == null || entry.getValue() == null) {
                throw new WaitResponseTimeoutException("未收到响应结果！！！");
            }
            return (RpcResponse) entry.getValue();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            ChannelPromiseHolder.removeEntry(txid);
        }
    }

    @Override
    public void stop() {
        shouldRetry.set(Boolean.FALSE);
        while (status.get() == ClientConnectionStatus.PENDING.getValue()) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException ignored) {
            }
        }
        if (status.compareAndSet(ClientConnectionStatus.STARTED.getValue(), ClientConnectionStatus.OFF.getValue())) {
            clear();
        }
    }

    private void clear() {
        try {
            connection.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            connection = null;
        }
        workerGroup.shutdownGracefully();
    }
}
